package nachos.network;
import nachos.machine.*;
import nachos.threads.*;
import java.util.Hashtable;
import java.util.LinkedList;
/**
 * A collection of message queues, one for each local port. A
 * <tt>PostOffice</tt> interacts directly with the network hardware. Because
 * of the network hardware, we are guaranteed that messages will never be
 * corrupted, but they might get lost.
 *
 * <p>
 * The post office uses a "postal worker" thread to wait for messages to arrive
 * from the network and to place them in the appropriate queues. This cannot
 * be done in the receive interrupt handler because each queue (implemented
 * with a <tt>SynchList</tt>) is protected by a lock.
 */
public class PostOffice {
	/**
	 * Allocate a new post office, using an array of <tt>SynchList</tt>s.
	 * Register the interrupt handlers with the network hardware and start the
	 * "postal worker" thread.
	 */
	public PostOffice(){
		messageReceived = new Semaphore(0);
		messageSent = new Semaphore(0);
		sendLock = new Lock();
		activeSockets = new Hashtable<Integer, LinkedList<NetSocket>>();

		Runnable receiveHandler = new Runnable() {
			public void run(){
				receiveInterrupt();
			}
		};
		Runnable sendHandler = new Runnable() {
			public void run(){
				sendInterrupt();
			}
		};
		Machine.networkLink().setInterruptHandlers(receiveHandler,
			sendHandler);

		KThread csend = new KThread(new Runnable() {
			public void run(){
				cycleSend();
			}
		}).setName("PO, sender");
		KThread t = new KThread(new Runnable() {
			public void run(){
				postalDelivery();
			}
		}).setName("PO, reciever");

		t.fork();
		csend.fork();
	}
	/**
	 * Wait for incoming messages, and then put them in the correct mailbox.
	 */
	private void postalDelivery(){
		while (true) {
			messageReceived.P();

			Packet p = Machine.networkLink().receive();

			MailMessage mail;

			try {
				mail = new MailMessage(p);
			} catch (MalformedPacketException e) {
				continue;
			}

			if(Lib.test(dbgNet)){
				System.out.println("delivering mail to port " + mail.dstPort
					+ ": " + mail);
			}

			// atomically add message to the mailbox and wake a waiting thread
			if(activeSockets.containsKey(mail.dstPort)){
				listlock.acquire();
				for (NetSocket s : activeSockets.get(mail.dstPort)) {
					s.recieve(mail);
				}
				listlock.release();
			}
		}
	}
	/**
	 * Called when a packet has arrived and can be dequeued from the network
	 * link.
	 */
	private void receiveInterrupt(){
		messageReceived.V();
	}
	/**
	 * Send a message to a mailbox on a remote machine.
	 */
	public void send(MailMessage mail){
		if(Lib.test(dbgNet)){
			System.out.println("sending mail: " + mail);
		}

		sendLock.acquire();

		Machine.networkLink().send(mail.packet);
		messageSent.P();

		sendLock.release();
	}
	private void cycleSend(){
		while (true) {
			NetKernel.timeout();
			System.out.flush();
			listlock.acquire();
			for (LinkedList<NetSocket> l : activeSockets.values()) {
				for (NetSocket s : l) {
					s.sendAll();
				}
			}
			listlock.release();
		}
	}
	/**
	 * Called when a packet has been sent and another can be queued to the
	 * network link. Note that this is called even if the previous packet was
	 * dropped.
	 */
	private void sendInterrupt(){
		messageSent.V();
	}
	public void rememberSocket(NetSocket sckt, int portnum){
		listlock.acquire();
		if(!activeSockets.containsKey(portnum)){
			activeSockets.put(portnum, new LinkedList<NetSocket>());
		}
		activeSockets.get(portnum).add(sckt);
		listlock.release();
	}
	public void forgetSocket(NetSocket sckt, int portnum){
		listlock.acquire();
		if(!activeSockets.containsKey(portnum)){
			listlock.release();
			return;
		}
		activeSockets.get(portnum).remove(sckt);
		if(activeSockets.get(portnum).isEmpty()){
			activeSockets.remove(portnum);
		}
		listlock.release();
	}
	public int newConnectionId(int portnum){
		listlock.acquire();
		if(!activeSockets.containsKey(portnum)){
			listlock.release();
			return 0;
		}
		listlock.release();
		return activeSockets.get(portnum).size();
	}
	public Hashtable<Integer, LinkedList<NetSocket>> activeSockets;
	private Semaphore messageReceived;	// V'd when a message can be dequeued
	private Semaphore messageSent;	// V'd when a message can be queued
	private Lock sendLock;
	private Lock listlock = new Lock();;
	private static final char dbgNet = 'n';
}
